// #pragma once
// #include <etcd/Client.hpp>
// #include <etcd/KeepAlive.hpp>
// #include <etcd/Response.hpp>
// #include <etcd/Watcher.hpp>
// #include <etcd/Value.hpp>
// #include <thread>
// #include <functional>
// #include "logger.hpp"

// namespace suye
// {
// // 服务注册客户端类
// class Registry
// {
// public:
//     using ptr = std::shared_ptr<Registry>;

//     Registry(const std::string& host)
//         : _client(std::make_shared<etcd::Client>(host))
//         , _keep_alive(_client->leasekeepalive(3).get())
//         , _lease_id(_keep_alive->Lease())
//     {}

//     bool registry(const std::string& key, const std::string& val)
//     {
//         auto resp = _client->put(key, val, _lease_id).get();
//         if(resp.is_ok() == false)
//         {
//             LOG_ERROR("注册数据失败: {}", resp.error_message());
//             return false;
//         }
//         return true;
//     }

//     ~Registry()
//     {
//         _keep_alive->Cancel();
//     }

// private:
//     std::shared_ptr<etcd::Client> _client;
//     std::shared_ptr<etcd::KeepAlive> _keep_alive;
//     u_int64_t _lease_id;
// };

// // 服务发现客户端类
// class Discovery
// {
// public:
//     using ptr = std::shared_ptr<Discovery>;
//     using NotifyCallback = std::function<void(std::string, std::string)>;   // 新增: 服务变更回调（新增/删除）

//     Discovery(const std::string& host, 
//             const std::string& basedir, 
//             const NotifyCallback& put_cb, 
//             const NotifyCallback& del_cb)
//         : _client(std::make_shared<etcd::Client>(host))
//         , _put_cb(put_cb)
//         , _del_cb(del_cb)
//     {
//         // 1.先进行服务发现，获取当前的所有服务信息
//         auto resp = _client->ls(basedir).get();
//         if(resp.is_ok() == false)
//         {
//             LOG_ERROR("获取键值对数据失败: {}", resp.error_message());
//         }
//         int sz = resp.keys().size();
//         for(int i = 0; i < sz; ++i)
//         {
//             if(_put_cb) 
//             {
//                 _put_cb(resp.key(i), resp.value(i).as_string());
//             }
//         }
//         // 2.然后进行事件监控，监控数据发生的改变并调用回调函数进行处理
//         _watcher = std::make_shared<etcd::Watcher>(*_client.get(), basedir, 
//             std::bind(&Discovery::Callback, this, std::placeholders::_1), true);
//     }

// private:
//     void Callback(const etcd::Response resp)
//     {
//         if(resp.is_ok() == false)
//         {
//             LOG_DEBUG("收到一个错误的事件通知: {}", resp.error_message());
//             return;
//         }
//         for(auto const& ev: resp.events())
//         {
//             if(ev.event_type() == etcd::Event::EventType::PUT)
//             {
//                 if(_put_cb)
//                 {
//                     _put_cb(ev.kv().key(), ev.kv().as_string());
//                 }
//                 LOG_DEBUG("服务信息新增: {} - {}", ev.kv().key(), ev.kv().as_string());
//             }
//             else if(ev.event_type() == etcd::Event::EventType::DELETE_)
//             {
//                 if(_del_cb)
//                 {
//                     _del_cb(ev.prev_kv().key(), ev.prev_kv().as_string());
//                 }
//                 LOG_DEBUG("服务信息删除: {} - {}", ev.prev_kv().key(), ev.prev_kv().as_string());
//             }
//         }
//     }

// private:
//     NotifyCallback _put_cb;
//     NotifyCallback _del_cb;
//     std::shared_ptr<etcd::Client> _client;
//     std::shared_ptr<etcd::Watcher> _watcher;
// };
// }

#pragma once
#include <etcd/Client.hpp>
#include <etcd/KeepAlive.hpp>
#include <etcd/Response.hpp>
#include <etcd/Watcher.hpp>
#include <etcd/Value.hpp>
#include <thread>
#include <functional>
#include <unordered_map>
#include <chrono>
#include <atomic>  // 新增：原子变量用于线程控制
#include "logger.hpp"

namespace suye
{
class Registry
{
public:
    using ptr = std::shared_ptr<Registry>;

    Registry(const std::string& host)
        : _client(std::make_shared<etcd::Client>(host))
        , _keep_alive(_client->leasekeepalive(3).get())
        , _lease_id(_keep_alive->Lease())
    {}

    bool registry(const std::string& key, const std::string& val)
    {
        auto resp = _client->put(key, val, _lease_id).get();
        if(resp.is_ok() == false)
        {
            LOG_ERROR("注册数据失败: {}", resp.error_message());
            return false;
        }
        return true;
    }

    ~Registry()
    {
        _keep_alive->Cancel();
    }

private:
    std::shared_ptr<etcd::Client> _client;
    std::shared_ptr<etcd::KeepAlive> _keep_alive;
    u_int64_t _lease_id;
};

// 服务发现客户端类（核心修改）
class Discovery
{
public:
    using ptr = std::shared_ptr<Discovery>;
    using NotifyCallback = std::function<void(std::string, std::string)>;  // 服务变更回调（新增/删除）

    Discovery(const std::string& host, 
            const std::string& basedir, 
            const NotifyCallback& put_cb, 
            const NotifyCallback& del_cb)
        : _client(std::make_shared<etcd::Client>(host))
        , _put_cb(put_cb)
        , _del_cb(del_cb)
        , _retry_count(0)
        , _last_sync_success(false)
        , _retry_timer_running(false)
    {
        // 1. 首次同步服务列表（并更新缓存）
        SyncServiceList(basedir);
        
        // 2. 启动etcd事件监听
        _watcher = std::make_shared<etcd::Watcher>(*_client.get(), basedir, 
            std::bind(&Discovery::Callback, this, std::placeholders::_1), true);
    }

    ~Discovery() {
        _retry_timer_running = false;  // 停止定时器线程
        if (_retry_thread.joinable()) {
            _retry_thread.join();  // 等待线程退出
        }
    }

private:
    // 核心逻辑：同步服务列表（含缓存更新）
    void SyncServiceList(const std::string& basedir) {
        auto resp = _client->ls(basedir).get();
        if (resp.is_ok()) {
            // 同步成功：更新缓存并触发回调
            std::unordered_map<std::string, std::string> new_services;
            for (int i = 0; i < resp.keys().size(); ++i) {
                std::string key = resp.key(i);
                std::string value = resp.value(i).as_string();
                new_services[key] = value;
                if (_put_cb) _put_cb(key, value);  // 触发新增回调
            }
            _cached_services = new_services;  // 更新缓存
            _last_sync_success = true;
            _retry_count = 0;
            LOG_INFO("etcd服务列表同步成功，缓存更新");
        } else {
            // 同步失败：使用缓存并启动重试
            if (_last_sync_success) {
                LOG_WARN("etcd同步失败，使用本地缓存（{}个服务）", _cached_services.size());
                for (const auto& [key, value] : _cached_services) {
                    if (_put_cb) _put_cb(key, value);  // 从缓存恢复服务
                }
            } else {
                LOG_ERROR("etcd初始化失败且无缓存可用");
            }
            _last_sync_success = false;
            StartRetryTimer(basedir);  // 启动重试
        }
    }

    // 重试定时器（每5秒重试，最多3次）
    void StartRetryTimer(const std::string& basedir) {
        if (_retry_count >= MAX_RETRY_TIMES) {
            LOG_ERROR("etcd重试次数已达上限（{}次），停止重试", MAX_RETRY_TIMES);
            return;
        }
        _retry_timer_running = true;
        _retry_thread = std::thread([this, basedir]() {
            while (_retry_timer_running && _retry_count < MAX_RETRY_TIMES) {
                std::this_thread::sleep_for(std::chrono::milliseconds(5000));  // 5秒间隔
                LOG_INFO("尝试第{}次重新连接etcd...", _retry_count + 1);
                SyncServiceList(basedir);  // 重新同步
                _retry_count++;
            }
        });
    }

    // etcd事件监听回调（新增缓存更新逻辑）
    void Callback(const etcd::Response resp) {
        if (resp.is_ok() == false) {
            LOG_DEBUG("收到错误事件通知: {}", resp.error_message());
            return;
        }
        // 更新缓存并触发业务回调
        for (const auto& ev : resp.events()) {
            if (ev.event_type() == etcd::Event::EventType::PUT) {
                _cached_services[ev.kv().key()] = ev.kv().as_string();  // 更新缓存
                if (_put_cb) _put_cb(ev.kv().key(), ev.kv().as_string());
            } else if (ev.event_type() == etcd::Event::EventType::DELETE_) {
                _cached_services.erase(ev.prev_kv().key());  // 更新缓存
                if (_del_cb) _del_cb(ev.prev_kv().key(), ev.prev_kv().as_string());
            }
        }
        LOG_DEBUG("etcd事件处理完成，当前缓存服务数: {}", _cached_services.size());
    }

private:
    NotifyCallback _put_cb;                // 服务新增回调（业务层处理）
    NotifyCallback _del_cb;                // 服务删除回调（业务层处理）
    std::shared_ptr<etcd::Client> _client; // etcd客户端
    std::shared_ptr<etcd::Watcher> _watcher; // etcd监听器
    std::unordered_map<std::string, std::string> _cached_services; // 服务缓存（键→值）
    int _retry_count;                      // 当前重试次数
    bool _last_sync_success;               // 最后一次同步是否成功
    static constexpr int MAX_RETRY_TIMES = 3; // 最大重试次数
    std::atomic<bool> _retry_timer_running;  // 新增：控制定时器线程运行状态
    std::thread _retry_thread;             // 新增：定时器线程
};
}